package bigdata.backend.meta.service;

import bigdata.backend.datasource.entity.Datasource;
import bigdata.backend.datasource.service.DatasourceService;
import bigdata.backend.datasource.utils.SchemaSyncException;
import bigdata.backend.meta.dto.*;
import bigdata.backend.meta.entity.DorisColumnMeta;
import bigdata.backend.meta.entity.DorisTableMeta;
import bigdata.backend.meta.utils.DorisTableSqlParser;
import bigdata.backend.utils.TableNameValidator;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.time.LocalDateTime;
import java.util.*;
import java.util.stream.Collectors;

/**
 * Doris 表元数据抓取、比对与双向同步服务（支持逻辑删除）
 * <p>
 * 核心能力：
 * 1. 抓取 Doris 表完整元数据并持久化；
 * 2. 比对本地元数据（仅 delete_flag='0'）与 Doris 实际结构，返回差异项；
 * 3. 支持两个方向的同步：
 * - syncFromDorisToLocal：拉取 Doris 变更 → 更新元数据（设 deleteFlag="1" 表示删除）
 * - syncFromLocalToDoris：推送本地变更 → 生成并执行 ALTER DDL
 * 4. 支持按字段粒度操作（单条差异项同步）
 * </p>
 */
@Slf4j
@Service
public class DorisMetaService {

    private static final ObjectMapper objectMapper = new ObjectMapper();

    @Autowired
    private DatasourceService datasourceService;

    @Autowired
    private DorisTableMetaService dorisTableMetaService;

    @Autowired
    private DorisColumnMetaService dorisColumnMetaService;
    
    

    @Transactional(rollbackFor = Exception.class)
    public DorisTableMeta saveDorisTableMeta(DorisTableWithColumns fullMeta) {
        DorisTableMeta tableMeta = fullMeta.getTableMeta();
        List<DorisColumnMeta> columns = fullMeta.getColumns();

        dorisTableMetaService.save(tableMeta);
        Long tableId = tableMeta.getId();

        if (columns != null && !columns.isEmpty()) {
            columns.forEach(col -> {
                col.setTableId(tableId);
                col.setDeleteFlag("0"); // 新增字段默认未删除
                col.setCreateTime(LocalDateTime.now());
                col.setUpdateTime(LocalDateTime.now());
            });
            dorisColumnMetaService.saveBatch(columns);
        }
        return tableMeta;
    }

    // ==================== 新增：差异比对 ====================

    /**
     * 比对本地元数据（仅未删除字段）与 Doris 实际字段结构，返回差异列表
     * <p>
     * 注意：此方法仅用于“检测差异”，不执行任何写操作。
     * </p>
     *
     * @param tableName       表名
     * @param storedTableMeta 本地已存储的表元数据（必须包含有效 ID）
     * @return 差异项列表
     */
    public List<DorisColumnDiffItem> compareTableColumns(String tableName, DorisTableMeta storedTableMeta) {
        Long datasourceId = storedTableMeta.getDatasourceId();
        Datasource ds = datasourceService.getById(datasourceId);
        // 1. 从 Doris 获取最新字段
        List<Map<String, Object>> liveRaw = datasourceService.getTableColumns(ds, tableName);
        if (liveRaw == null || liveRaw.isEmpty()) {
            throw new SchemaSyncException("Doris 表不存在或无字段: " + tableName);
        }
        String createSql = datasourceService.getCreateTableSql(ds, tableName);
        DorisTableStruct liveStruct = DorisTableSqlParser.parse(createSql);
        List<DorisColumnDto> liveColumns = buildColumnDtos(liveRaw, liveStruct);

        // 2. 从本地加载字段（仅未删除：delete_flag = "0"）
        List<DorisColumnMeta> storedColumns = dorisColumnMetaService.list(
                new QueryWrapper<DorisColumnMeta>()
                        .eq("table_id", storedTableMeta.getId())
                        .eq("delete_flag", "0")
        );

        // 3. 构建快速查找 Map
        Map<String, DorisColumnDto> liveMap = liveColumns.stream()
                .collect(Collectors.toMap(DorisColumnDto::getName, c -> c));
        Map<String, DorisColumnMeta> storedMap = storedColumns.stream()
                .collect(Collectors.toMap(DorisColumnMeta::getColumnName, c -> c));

        // 4. 比对所有字段名（并集）
        Set<String> allNames = new HashSet<>();
        allNames.addAll(liveMap.keySet());
        allNames.addAll(storedMap.keySet());

        List<DorisColumnDiffItem> diffs = new ArrayList<>();
        for (String name : allNames) {
            DorisColumnDto live = liveMap.get(name);
            DorisColumnMeta stored = storedMap.get(name);

            if (stored == null) {
                // Doris 有，本地无（或已逻辑删除）→ ADD
                DorisColumnDiffItem item = new DorisColumnDiffItem();
                item.setType(DorisColumnDiffItem.DiffType.ADD);
                item.setColumnName(name);
                item.setLiveMeta(live);
                diffs.add(item);
            } else if (live == null) {
                // 本地有（未删除），Doris 无 → DELETE
                DorisColumnDiffItem item = new DorisColumnDiffItem();
                item.setType(DorisColumnDiffItem.DiffType.DELETE);
                item.setColumnName(name);
                item.setOldMeta(stored);
                diffs.add(item);
            } else if (!isColumnEqual(stored, live)) {
                // 都存在但不同 → UPDATE
                DorisColumnDiffItem item = new DorisColumnDiffItem();
                item.setType(DorisColumnDiffItem.DiffType.UPDATE);
                item.setColumnName(name);
                item.setOldMeta(stored);
                item.setLiveMeta(live);
                diffs.add(item);
            }
        }
        return diffs;
    }


    // ==================== 新增：单字段粒度同步 ====================

    /**
     * 将 Doris 的变更拉取到本地元数据
     * - ADD 差异（Doris有，本地无）→ 新增元数据
     * - DELETE 差异（本地有，Doris无）→ 逻辑删除元数据
     * - UPDATE 差异 → 更新元数据
     */
    @Transactional(rollbackFor = Exception.class)
    public void syncSingleDiffFromDorisToLocal(DorisColumnDiffItem diff, DorisTableMeta tableMeta) {
        LocalDateTime now = LocalDateTime.now();
        if (diff.getType() == DorisColumnDiffItem.DiffType.ADD) {
            // Doris有，本地无 → 新增元数据
            DorisColumnMeta meta = convertToMeta(diff.getLiveMeta(), tableMeta.getId());
            meta.setDeleteFlag("0");
            meta.setCreateTime(now);
            meta.setUpdateTime(now);
            dorisColumnMetaService.save(meta);
        } else if (diff.getType() == DorisColumnDiffItem.DiffType.DELETE) {
            // 本地有，Doris无 → 逻辑删除
            DorisColumnMeta meta = diff.getOldMeta();
            meta.setDeleteFlag("1");
            meta.setUpdateTime(now);
            dorisColumnMetaService.updateById(meta);
        } else if (diff.getType() == DorisColumnDiffItem.DiffType.UPDATE) {
            // 两边都有，但不同 → 更新元数据
            DorisColumnMeta updated = convertToMeta(diff.getLiveMeta(), tableMeta.getId());
            updated.setId(diff.getOldMeta().getId());
            updated.setDeleteFlag("0");
            updated.setCreateTime(diff.getOldMeta().getCreateTime());
            updated.setUpdateTime(now);
            dorisColumnMetaService.updateById(updated);
        } else {
            throw new IllegalArgumentException("不支持的差异类型: " + diff.getType());
        }
    }


    // ==================== 工具方法：构建 DorisColumnDto ====================
    // 因为有自有属性所以不能统一

    private List<DorisColumnDto> buildColumnDtos(List<Map<String, Object>> basicColumns, DorisTableStruct struct) {
        List<DorisColumnDto> dtos = new ArrayList<>();
        for (int i = 0; i < basicColumns.size(); i++) {
            Map<String, Object> row = basicColumns.get(i);
            DorisColumnDto dto = new DorisColumnDto();
            dto.setName((String) row.get("name"));
            dto.setType((String) row.get("type"));
            dto.setNullable((Boolean) row.getOrDefault("nullable", true));
            dto.setDefaultValue((String) row.get("default"));
            dto.setComment((String) row.getOrDefault("comment", ""));
            dto.setOrdinalPosition(i + 1);
            dto.setIsKey(struct.getKeyColumns() != null && struct.getKeyColumns().contains(dto.getName()));
            dto.setIsPartitionColumn(
                    struct.getPartitionColumn() != null &&
                            struct.getPartitionColumn().equals(dto.getName())
            );
            dtos.add(dto);
        }
        return dtos;
    }

    // ==================== 工具方法：字段等值判断 ====================

    private boolean isColumnEqual(DorisColumnMeta stored, DorisColumnDto live) {
        if (stored == null || live == null) return false;
        return Objects.equals(stored.getColumnName(), live.getName()) &&
                Objects.equals(normalizeType(stored.getColumnType()), normalizeType(live.getType())) &&
                Objects.equals(stored.getNullable(), live.getNullable()) &&
                Objects.equals(stored.getDefaultValue(), live.getDefaultValue()) &&
                Objects.equals(stored.getColumnComment(), live.getComment()) &&
                Objects.equals(stored.getIsKey(), live.getIsKey()) &&
                Objects.equals(stored.getIsPartitionColumn(), live.getIsPartitionColumn());
    }

    private String normalizeType(String type) {
        if (type == null) return null;
        String lower = type.toLowerCase();
        if (lower.matches("^(bigint|int|integer|smallint|tinyint)\\(\\d+\\)$")) {
            return lower.substring(0, lower.indexOf('('));
        }
        return type;
    }

    // ==================== 工具方法：DDL 生成 ====================

    // ==================== 工具方法：实体转换 ====================

    private DorisColumnMeta convertToMeta(DorisColumnDto dto, Long tableId) {
        DorisColumnMeta meta = new DorisColumnMeta();
        meta.setTableId(tableId);
        meta.setColumnName(dto.getName());
        meta.setColumnType(dto.getType());
        meta.setNullable(dto.getNullable());
        meta.setDefaultValue(dto.getDefaultValue());
        meta.setColumnComment(dto.getComment());
        meta.setOrdinalPosition(dto.getOrdinalPosition());
        meta.setIsKey(dto.getIsKey());
        meta.setIsPartitionColumn(dto.getIsPartitionColumn());
        meta.setAggregationType(dto.getAggregationType());
        return meta;
    }

    // ==================== 原有内部抓取逻辑（保持不变） ====================

    public DorisTableWithColumns getDorisTableAndColumnMeta(Long datasourceId, String tableName) {
        Datasource ds = datasourceService.getById(datasourceId);
        tableName = TableNameValidator.validate(tableName.trim());
        List<Map<String, Object>> basicColumns = datasourceService.getTableColumns(ds, tableName);
        if (basicColumns == null || basicColumns.isEmpty()) {
            throw new SchemaSyncException("Doris 表 `" + tableName + "` 不存在或无字段定义");
        }
        String createSql = datasourceService.getCreateTableSql(ds, tableName);
        if (createSql == null || createSql.trim().isEmpty()) {
            throw new SchemaSyncException("无法获取 Doris 表 `" + tableName + "` 的建表语句");
        }
        DorisTableStruct struct = DorisTableSqlParser.parse(createSql);
        List<DorisColumnMeta> columnMetas = buildColumnMetas(basicColumns, struct);
        DorisTableMeta tableMeta = buildTableMeta(ds, tableName, struct);
        DorisTableWithColumns result = new DorisTableWithColumns();
        result.setTableMeta(tableMeta);
        result.setColumns(columnMetas);
        return result;
    }

    private List<DorisColumnMeta> buildColumnMetas(List<Map<String, Object>> basicColumns, DorisTableStruct struct) {
        List<DorisColumnMeta> columns = new ArrayList<>();
        for (int i = 0; i < basicColumns.size(); i++) {
            Map<String, Object> row = basicColumns.get(i);
            String name = (String) row.get("name");
            String type = (String) row.get("type");
            Boolean nullable = (Boolean) row.get("nullable");
            String defaultValue = (String) row.get("default");
            String comment = (String) row.get("comment");

            DorisColumnDto dto = new DorisColumnDto();
            dto.setName(name);
            dto.setType(type);
            dto.setNullable(nullable != null ? nullable : true);
            dto.setDefaultValue(defaultValue);
            dto.setComment(comment != null ? comment : "");
            dto.setOrdinalPosition(i + 1);
            dto.setIsKey(struct.getKeyColumns() != null && struct.getKeyColumns().contains(name));
            dto.setIsPartitionColumn(
                    struct.getPartitionColumn() != null &&
                            struct.getPartitionColumn().equals(name)
            );

            DorisColumnMeta entity = new DorisColumnMeta();
            entity.setColumnName(dto.getName());
            entity.setColumnType(dto.getType());
            entity.setNullable(dto.getNullable());
            entity.setDefaultValue(dto.getDefaultValue());
            entity.setColumnComment(dto.getComment());
            entity.setOrdinalPosition(dto.getOrdinalPosition());
            entity.setIsKey(dto.getIsKey());
            entity.setIsPartitionColumn(dto.getIsPartitionColumn());
            entity.setAggregationType(dto.getAggregationType());
            entity.setDeleteFlag("0"); // 新建默认未删除
            entity.setCreateTime(LocalDateTime.now());
            entity.setUpdateTime(LocalDateTime.now());
            columns.add(entity);
        }
        return columns;
    }

    private DorisTableMeta buildTableMeta(Datasource ds, String tableName, DorisTableStruct struct) {
        DorisTableMeta meta = new DorisTableMeta();
        meta.setDatasourceId(ds.getId());
        meta.setTableName(tableName);
        String tableComment = datasourceService.getTableComment(ds, tableName);
        meta.setTableComment(tableComment != null ? tableComment : "");
        meta.setTableType(struct.getKeyType());
        meta.setBucketNum(struct.getBucketNum());
        meta.setReplicationNum(struct.getReplicationNum());
        meta.setDeleteFlag("0"); // 新建表默认未删除

        try {
            meta.setKeyColumns(objectMapper.writeValueAsString(struct.getKeyColumns()));
            meta.setBucketColumns(objectMapper.writeValueAsString(struct.getBucketColumns()));
            meta.setPartitionInfo(buildPartitionInfoJson(struct));
            meta.setProperties(objectMapper.writeValueAsString(struct.getProperties()));
        } catch (JsonProcessingException e) {
            throw new SchemaSyncException("JSON 序列化失败", e);
        }

        LocalDateTime now = LocalDateTime.now();
        meta.setCreateTime(now);
        meta.setUpdateTime(now);
        return meta;
    }

    private String buildPartitionInfoJson(DorisTableStruct struct) throws JsonProcessingException {
        if (struct.getPartitionColumn() == null) {
            return objectMapper.writeValueAsString(Collections.emptyMap());
        }
        Map<String, Object> partitionInfo = new HashMap<>();
        partitionInfo.put("column", struct.getPartitionColumn());
        partitionInfo.put("values", struct.getPartitionValues());
        return objectMapper.writeValueAsString(partitionInfo);
    }
}